package com.rocketmq.demo;

import com.rocketmq.constants.MqConstants;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.junit.Test;

/**
 * @ClassName COnewayTest
 * @Author Patrick Star
 * @Date 2025/7/27 16:22
 */
public class COnewayTest {

    @Test
    public void OnewayProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("oneway-producer-group");
        producer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        producer.start();
        Message message = new Message("onewayTopic", "我是一个单向消息".getBytes());
        producer.sendOneway(message);
        System.out.println("发送成功");
        producer.shutdown();
    }

    @Test
    public void OnewayConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("oneway-consumer-group");
        consumer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);

        consumer.subscribe("onewayTopic", "*");
        consumer.registerMessageListener((MessageListenerConcurrently) (message, context) -> {
            System.out.println("我是一个消费者");
            System.out.println(message);
            System.out.println(new String(message.get(0).getBody()));
            System.out.println(context);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.start();
        System.in.read();
    }
}
